package com.ds.lens.collector.receive.handler;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ds.lens.collector.receive.kafka.KafkaProducer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

/**
 * Description:
 *
 * @author WeiShaoying
 * @date 2020-04-01
 */
@Slf4j
@Service
public class EventsHandler extends AbstractHandler {

    @Autowired
    private KafkaProducer kafkaProducer;

    @Value("${spring.kafka.events.topic}")
    private String eventsTopic;

    @Override
    protected HandlerResultWrapper handleData(String jsonStr, RequestInfo requestInfo) {
        try {

            JSONObject jsonObject = JSON.parseObject(jsonStr);
            jsonObject.put("appKey", requestInfo.getAppKey());
            kafkaProducer.sendMessage(eventsTopic, jsonObject.toJSONString());
            HandlerResultWrapper handlerResultWrapper = new HandlerResultWrapper();
            handlerResultWrapper.setResult("OK");
            return handlerResultWrapper;
        } catch (Exception e) {
            log.warn("events handle error. Caused:{}.", e.getMessage());
        }
        return null;
    }
}
